package streaming.api.transform;

import streaming.api.beans.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 合流
 * connect 和 CoMap
 * 只能合两条流，数据类型可不一致
 */
public class TransformTest5 {

    public static void main(String[] args) throws Exception {
        /**运行环境*/
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStream<String> inputStream = env.readTextFile("D:\\IdeaProjects\\springboot-flink-1\\flinkTutorial\\src\\main\\resources\\sensor.txt");
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0],new Long(fields[1]), new Double(fields[2]));
        });
        /**1、定义OutputTag*/
        OutputTag<SensorReading> sideOutputTag = new OutputTag<SensorReading>("SensorReadingOutputTag"){};
        /**2、在ProcessFunction中处理主流和分流*/
        SingleOutputStreamOperator<SensorReading> processedStream = dataStream.process(new ProcessFunction<SensorReading, SensorReading>() {
            @Override
            public void processElement(SensorReading sensorReading, Context context, Collector<SensorReading> collector) throws Exception {
                //侧流-只输出特定数据
                if (sensorReading.getTemperature() > 30) {
                    context.output(sideOutputTag, sensorReading);
                }
                //主流
                else {
                    collector.collect(sensorReading);
                }
            }
        });
        //获取主流
        processedStream.print("main:");
        //获取侧流
//        processedStream.getSideOutput(sideOutputTag).print("side:");
        DataStream<Tuple2<String, Double>> warningStream = processedStream.getSideOutput(sideOutputTag).map(new MapFunction<SensorReading, Tuple2<String, Double>>() {
            @Override
            public Tuple2<String, Double> map(SensorReading value) throws Exception {
                return new Tuple2<>(value.getId(), value.getTemperature());
            }
        });
        warningStream.print("warning:");

        //合并流
        ConnectedStreams<SensorReading, Tuple2<String, Double>> connectedStream = processedStream.connect(warningStream);
        SingleOutputStreamOperator<Tuple2<String, String>> resultStream = connectedStream.map(new CoMapFunction<SensorReading, Tuple2<String, Double>, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map1(SensorReading val1) throws Exception {
                return  new Tuple2<>(val1.getId(), String.valueOf(val1.getTemperature()));
            }
            @Override
            public Tuple2<String, String> map2(Tuple2<String, Double> val2) throws Exception {
                return new Tuple2<>(val2.f0, String.valueOf(val2.f1));
            }
        });
        resultStream.print("connect:");

//        /**输入数据源source1*/
//        DataStream<Tuple3<String, String, String>> source1 = env.fromElements(new Tuple3<>("productID1", "click", "user_1"));
//        /**输入数据源source2*/
//        DataStream<String> source2 = env.fromElements("productID3:click:user_1", "productID3:browse:user_2");
//        /**1、合并流*/
//        ConnectedStreams<Tuple3<String, String, String>, String> connectedStream = source1.connect(source2);
//        /**2、用CoMap处理合并后的流*/
//        SingleOutputStreamOperator<Tuple3<String, String, String>> resultStream = connectedStream.map(new CoMapFunction<Tuple3<String, String, String>, String, Tuple3<String, String, String>>() {
//            //定义第一个流的处理逻辑
//            @Override
//            public Tuple3<String, String, String> map1(Tuple3<String, String, String> value) throws Exception {
//                return new Tuple3<>(value.f0, value.f1, value.f2);
//            }
//            //定义第二个流的处理逻辑
//            @Override
//            public Tuple3<String, String, String> map2(String value) throws Exception {
//                String[] valueSplit = value.split(":");
//                return new Tuple3<>(valueSplit[0], valueSplit[1], valueSplit[2]);
//            }
//        });
//        resultStream.print();

        env.execute();
    }
}
